package org.example.nettyrpc;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import org.example.nettyrpc.common.IpPortTuple;
import org.example.nettyrpc.common.RequestArgs;
import org.example.nettyrpc.common.ResponseReply;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;

public class RpcHost {
    private final EndpointBase endpoint;
    private final IpPortTuple ipPortTuple;
    private volatile boolean hasConnected = false;

    public RpcHost(EndpointBase endpoint, IpPortTuple ipPortTuple) {
        this.endpoint = endpoint;
        this.ipPortTuple = ipPortTuple;
    }

    public void waitForConnected() {
        while (!hasConnected) {
            Thread.yield();
        }
    }

    public void run() throws Exception {
        NioEventLoopGroup boosG = new NioEventLoopGroup();
        NioEventLoopGroup workG = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(boosG, workG)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) {
                            socketChannel.pipeline().addLast(new ObjectDecoder(1024 * 1024,
                                    ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
                            socketChannel.pipeline().addLast(new ObjectEncoder());
                            socketChannel.pipeline().addLast(new ReceiveHandler(endpoint));
                        }

                    })
                    .option(ChannelOption.SO_BACKLOG, 1024);

            ChannelFuture f = b.bind(ipPortTuple.getPort()).sync();
            hasConnected = true;
            f.channel().closeFuture().sync();
        } finally {
            workG.shutdownGracefully();
            boosG.shutdownGracefully();
        }
    }

    private static class ReceiveHandler extends ChannelInboundHandlerAdapter {
        final EndpointBase endpoint;

        public ReceiveHandler(EndpointBase endpoint) {
            this.endpoint = endpoint;
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
            if (msg instanceof RequestArgs) {
                RequestArgs args = (RequestArgs) msg;
                Object[] parameters = args.getParameters();

                Object ret = null;
                if (parameters == null || parameters.length == 0) {
                    try {
                        Method method = endpoint.getClass().getMethod(args.getMethod());
                        ret = method.invoke(endpoint);
                        if (ret == null) {
                            // 无返回时，返回空字符串
                            ret = "";
                        }
                    } catch (Exception ex) {
                        ret = "invoke failed ex:" + ex;
                    }
                } else {
                    Class<?>[] parameterTypes = new Class<?>[parameters.length];
                    for (int i = 0; i < parameters.length; i++) {
                        if (parameters[i] == null) {
                            parameterTypes[i] = Object.class;
                        } else {
                            parameterTypes[i] = parameters[i].getClass();
                        }
                    }
                    try {
                        Method method = endpoint.getClass().getMethod(args.getMethod(), parameterTypes);
                        ret = method.invoke(endpoint, args.getParameters());
                    } catch (Exception ex) {
                        ret = "invoke failed ex:" + ex;
                    }
                }


                ResponseReply reply = new ResponseReply();
                reply.setSeq(args.getSeq());
                reply.setRes(ret);

                ctx.writeAndFlush(reply);
            } else {
                System.out.println("unknown request type");
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace();
            ctx.close();
        }
    }
}
